package knowlwdge_graph.knowledge_etl.yago_etl;

import org.slf4j.LoggerFactory;

import java.util.ArrayList;
import java.util.List;

/**
 * Created by julianzliu on 4/15/2017.
 */
public class TransformerF2O {
    /*
    transformer of postgres to orientdb
     */

    static final org.slf4j.Logger logger = LoggerFactory.getLogger(TransformerF2O.class);

    private ExtractorTSVFiles extractorTSVFiles;
    private LoderOrientDB loderOrientDB;

    public TransformerF2O(){
        this.extractorTSVFiles = new ExtractorTSVFiles();
        this.loderOrientDB = new LoderOrientDB();
    }


    public RowYagoTriple transform(RowYagoTriple row){
        //this.loderOrientDB.saveTripleTuple( row );
        return row;
    }


    public void saveMultipleRows(List<RowYagoTriple> rows) {
        for (RowYagoTriple row : rows) {
            RowYagoTriple turple = transform(row);
                /*
                多线程可能会冲突，需要多次重试机制
                 */
            int retry = 3;
            while (retry > 0) {
                try {
                    this.loderOrientDB.saveTripleTuple(turple);
                    break;
                } catch (Exception e) {
                    if (retry > 0) retry -= 1;
                    else {
                        logger.error("[Retry Failed!] msg = \n{}", e.getMessage());
                        try {
                            Thread.sleep(1000);
                        } catch (Exception e2) {
                            logger.error("[Sleep Failed!] msg = \n{}", e2.getMessage());
                        }
                    }
                }
            }

        }
    }


    public void traversalAll(int begin, int end, List<String> files){

        for(int i = begin; i< Math.min(end, files.size()) ; i++){
            String file = files.get(i);
            logger.info("\n\n\n[traversalAll] transform file = {}" + file);

            List<RowYagoTriple> rows = this.extractorTSVFiles.getAllTripleByTSVParser(file);

            saveMultipleRows(rows);
        }
    }


    public static class TransformerThread implements Runnable {

        private TransformerF2O transformer = new TransformerF2O();
        private int begin;
        private int end;
        private List<String> files;

        public TransformerThread(int begin, int end, List<String> files){
            super();
            this.begin = begin;
            this.end = end;
            this.files = files;
            logger.info("\n\n[TransformerThread] begin = {}, end = {}, total = {}\n\n", begin, end, files.size() );
        }

        public void run(){
            this.transformer.traversalAll(this.begin, this.end, this.files);
        }
    }


    /*****************************************************************
     * MAIN
     */
    public static void main(String[] args){
        //int batch_size = Integer.parseInt(args[0]);
        List<String> files = new ArrayList<>();
        files.add("/home/gammaliu/Datas/yago/yago3/copy_to_postgres/yagoSimpleTypesaa");
        TransformerF2O transformer = new TransformerF2O();
        transformer.traversalAll(0, 100000000, files);
    }
}
